package network;

import com.google.common.base.Throwables;
import utils.Closeables;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;

/**
 * 监听客户端的socket建立连接请求，然后在processor中对读写事件进行分发处理
 * **/
public class Accepter extends AbstractAccepter {

    private int port;
    private Processor processor;

    public Accepter(int port, Processor processor) {
        this.port = port;
        this.processor = processor;
    }

    //    @Override
    public void run() {
        ServerSocketChannel serverSocketChannel = null;
        try {
            serverSocketChannel = ServerSocketChannel.open();
            //设置为非阻塞
            serverSocketChannel.configureBlocking(false);
            serverSocketChannel.socket().bind(new InetSocketAddress(port));
            //监听所有已连接事件
            serverSocketChannel.register(getSelector(), SelectionKey.OP_ACCEPT);
        } catch (IOException e) {
            logger.error(e);
        }

        logger.info("waiting connection on port: {}", port);

        //调用countdown()，使得startupLatch == 0
        startupComplete();

        while (isRunning()) {
            int ready;
            try {
                ready = getSelector().select(500L);
            } catch (IOException e) {
                throw new IllegalStateException(e);
            }
            if (ready <= 0) {
                continue;
            }
            Iterator<SelectionKey> iterator = getSelector().selectedKeys().iterator();
            while (iterator.hasNext() && isRunning()) {
                try {
                    SelectionKey key = iterator.next();
                    iterator.remove();
                    //事件是否是已连接事件
                    if (key.isAcceptable()) {
                        accept(key, processor);
                    }
                } catch (Throwable t) {
                    logger.error("Error in acceptor {}", Throwables.getStackTraceAsString(t));
                }
            }
        }

        //run over
        logger.info("Closing server socket and selector.");
        Closeables.closeQuietly(serverSocketChannel);
        Closeables.closeQuietly(getSelector());
        shutdownComplete();
    }

    private void accept(SelectionKey key, Processor processor) throws IOException {
        ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();
        serverSocketChannel.socket().setReceiveBufferSize(1024 * 1024);

        SocketChannel socketChannel = serverSocketChannel.accept();
        socketChannel.configureBlocking(false);
        socketChannel.socket().setTcpNoDelay(true);
        socketChannel.socket().setSendBufferSize(1024 * 1024);

        processor.accept(socketChannel);
    }
}